
package il.technion.ewolf.beehive.op;

import static ch.lambdaj.Lambda.on;
import static ch.lambdaj.Lambda.sort;
import il.technion.ewolf.beehive.cache.DataCollector;
import il.technion.ewolf.beehive.cache.Storage;
import il.technion.ewolf.kbr.Key;
import il.technion.ewolf.kbr.KeyComparator;
import il.technion.ewolf.kbr.Node;
import il.technion.ewolf.kbr.concurrent.CompletionHandler;
import il.technion.ewolf.kbr.openkad.bucket.KBuckets;
import il.technion.ewolf.kbr.openkad.cache.KadCache;
import il.technion.ewolf.kbr.openkad.msg.FindValueRequest;
import il.technion.ewolf.kbr.openkad.msg.FindValueResponse;
import il.technion.ewolf.kbr.openkad.msg.KadMessage;
import il.technion.ewolf.kbr.openkad.net.MessageDispatcher;
import il.technion.ewolf.kbr.openkad.net.filter.IdMessageFilter;
import il.technion.ewolf.kbr.openkad.net.filter.TypeMessageFilter;

import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.name.Named;

/**
 * Find value operation where the value is a find node operation results
 * @author Ligal ligal2@gmail.com
 * @author Hen senderhen@gmail.com
 */
public class BeeHiveFindValueOperation extends MyFindValueOperation implements CompletionHandler<KadMessage, Node> {

	// state
	private List<Node> knownClosestNodes;
	private Serializable returnedValue;
	private Key key;
	private final Set<Node> alreadyQueried;
	private final Set<Node> querying;
	private int nrQueried;
	private Collection<Node> bootstrap = Collections.emptyList();
	private boolean gotStoredResult = false;
	
	// dependencies
	private final Provider<FindValueRequest> findValueRequestProvider;
	private final Provider<MessageDispatcher<Node>> msgDispatcherProvider;
	private final int kBucketSize;
	private final KBuckets kBuckets;
	private final Node localNode;
	private final KadCache cache;
	private final Storage valueCache;
	private final DataCollector statistics;
	private AtomicInteger nrNumberOfMsg;
	
	//for computing how much nodes we pass to receive value
	private HashMap<Key, List<Key>> searchPath; //node to who sent him to me 
	private AtomicInteger nrFindvalueHops;
	
	@Inject
	BeeHiveFindValueOperation(
			@Named("openkad.local.node") Node localNode,
			@Named("openkad.bucket.kbuckets.maxsize") int kBucketSize,
			@Named("openkad.testing.nrFindValueHops") AtomicInteger findvalueHops,
			@Named("openkad.testing.nRepMessages") AtomicInteger numberOfMsg,
			Provider<FindValueRequest> findValueRequestProvider,
			Provider<MessageDispatcher<Node>> msgDispatcherProvider,
			KBuckets kBuckets,
			KadCache cache,
			DataCollector collector,
			Storage p_extandCache) {
		this.localNode = localNode;
		this.kBucketSize = kBucketSize;
		this.kBuckets = kBuckets;
		this.findValueRequestProvider = findValueRequestProvider;
		this.msgDispatcherProvider = msgDispatcherProvider;
		this.cache = cache;
		this.valueCache = p_extandCache;
		this.statistics = collector;
		this.nrNumberOfMsg = numberOfMsg;
		
		alreadyQueried = new HashSet<Node>();
		querying = new HashSet<Node>();
		returnedValue = null;
		searchPath = new HashMap<Key, List<Key>>();
		nrFindvalueHops = findvalueHops;
	}
	
	/**
	 * Sets the key that we search the value of.
	 * @param key
	 * @return the operation
	 */
	public BeeHiveFindValueOperation setKey(Key key) {
		this.key = key;
		return this;
	}
	
	/**
	 * @return the number of nodes queried until the value returned 
	 */
	public int getNrQueried() {
		return nrQueried;
	}
	
	
	public BeeHiveFindValueOperation setBootstrap(Collection<Node> bootstrap) {
		this.bootstrap = bootstrap;
		return this;
	}
	
	/**
	 * @return Gets a node that has not yet queried
	 */
	private synchronized Node takeUnqueried() {
		for (int i=0; i < knownClosestNodes.size(); ++i) {
			Node n = knownClosestNodes.get(i);
			if (!querying.contains(n) && !alreadyQueried.contains(n)) {
				querying.add(n);
				return n;
			}
		}
		return null;
	}
	
	/***
	 * @return If there is more nodes to query
	 */
	private boolean hasMoreToQuery() {
		return !querying.isEmpty() || !alreadyQueried.containsAll(knownClosestNodes);
	}
	
	/***
	 * Send a message of 'FindValueRequest' to other node.
	 * @param to The node to send the message to
	 */
	private void sendFindValue(Node to) {
		FindValueRequest findValueRequest = findValueRequestProvider.get()
			.setKey(key)
			.setFlag(this.replication);
		
		msgDispatcherProvider.get()
			.addFilter(new IdMessageFilter(findValueRequest.getId()))
			.addFilter(new TypeMessageFilter(FindValueResponse.class))
			.setConsumable(true)
			.setCallback(to, this)
			.send(to, findValueRequest);
		
		if (this.replication)
			nrNumberOfMsg.incrementAndGet();
	}
	
	@Override
	public Serializable doFindValue() {
		//add to the statistics
		if (!this.replication)
			statistics.increse(key);
		
		//check in the node
		Serializable result = valueCache.search(key);
		if (result != null){
			return result;
		}
		
		//start searching in other nodes
		List<Node> cacheResults = cache.search(key);
		if (cacheResults != null){
			for (Node n : cacheResults){
				sendFindValue(n);
				querying.add(n);
			}
		}
			
		
		knownClosestNodes = kBuckets.getClosestNodesByKey(key, kBucketSize);
		knownClosestNodes.add(localNode);
		bootstrap.removeAll(knownClosestNodes);
		knownClosestNodes.addAll(bootstrap);
		alreadyQueried.add(localNode);
		KeyComparator keyComparator = new KeyComparator(key);
		
		do {
			synchronized(this) {
				knownClosestNodes = sort(knownClosestNodes, on(Node.class).getKey(), keyComparator);
				if (knownClosestNodes.size() >= kBucketSize)
					knownClosestNodes.subList(kBucketSize, knownClosestNodes.size()).clear();
				
				if (gotStoredResult)
					break;
				
				if (!hasMoreToQuery())
					break;
			}
			
			Node n = takeUnqueried();
			
			if (n != null) {
				sendFindValue(n);
			} else {
				synchronized (this) {
					if (!querying.isEmpty()) {
						try {
							wait();
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				}
			}
			
		} while (true);

		knownClosestNodes = Collections.unmodifiableList(knownClosestNodes);
		cache.insert(key, knownClosestNodes);
		
		synchronized (this) {
			nrQueried = alreadyQueried.size()-1 + querying.size();
		}
		
		return returnedValue;
		
	}
	
	@Override
	public synchronized void completed(KadMessage msg, Node n) {
		notifyAll();
		querying.remove(n);
		alreadyQueried.add(n);
		
		if (gotStoredResult)
			return;
		
		boolean isContainValue = ((FindValueResponse)msg).isContainingResult();
		if (isContainValue){
			Serializable valueReturned = ((FindValueResponse)msg).getValue();
			returnedValue = valueReturned;
			
			if (((FindValueResponse)msg).isStoredResults())
				gotStoredResult = true;
			
			try {
				nrFindvalueHops.addAndGet(computeDistance(((FindValueResponse)msg).getSrc().getKey(), new LinkedList<Key>()).get());
			}
			catch (Exception e){
				nrFindvalueHops.addAndGet(1);
			}
		}
		else {
			List<Node> nodes = ((FindValueResponse)msg).getNodes();
			if (nodes != null)
			{
				for (Node node : nodes){
					List<Key> value = new LinkedList<Key>();
					if (!node.getKey().equals(((FindValueResponse)msg).getSrc().getKey())){
						if (searchPath.containsKey(node.getKey())){
							value = searchPath.get(node.getKey());
							value.add(((FindValueResponse)msg).getSrc().getKey());
						}
						else 
						{
							value.add(((FindValueResponse)msg).getSrc().getKey());
							searchPath.put(node.getKey(), value);
						}
					}
				}
				nodes.removeAll(querying);
				nodes.removeAll(alreadyQueried);
				if (knownClosestNodes == null)
					knownClosestNodes = new LinkedList<Node>();
				nodes.removeAll(knownClosestNodes);
				
				try{
					knownClosestNodes.addAll(nodes);
				} catch (Exception e){
					//nothing
				}
			}
		}
		
	}
	
	/**
	 * Counts the distance of requests between the destination node to me
	 * @param dest
	 * @param previous
	 * @return
	 */
	private AtomicInteger computeDistance(Key dest, List<Key> previous) {
		AtomicInteger count = new AtomicInteger(Integer.MAX_VALUE);
		List<Key> path = new LinkedList<Key>();
		path.addAll(previous);
		path.add(dest);
		if(searchPath.containsKey(dest)){
			List<Key> result = searchPath.get(dest);
			for (Key key : result){
				if (!path.contains(key)) //prevent loop in the path
				{
					AtomicInteger temp = computeDistance(key, path);
					if (temp.get() <= count.get()){
						count.set(temp.get());
					}
				}
			}
			count.incrementAndGet();
		}
		else {
			count = new AtomicInteger(1);
		}
		
		return count;
	}

	@Override
	public synchronized void failed(Throwable exc, Node n) {
		notifyAll();
		querying.remove(n);
		alreadyQueried.add(n);
		
	}
}
